/**
 * 
 */
package com.nce.protocol.rtp;

import java.io.IOException;
import java.util.Vector;

import com.nce.protocol.rtp.RTPHeader;

import javax.media.Format;
import javax.media.protocol.PullBufferStream;
import javax.media.Buffer;


import com.sun.media.protocol.BasicSourceStream;



/**
 * @author domenico
 *
 */
public class ParticipantRTPStream extends BasicSourceStream implements PullBufferStream {

	private Format format = null;
	private Vector<byte[]> queue = null;
	//private RTPHeader nextHeader = null;
	
	private Object sync = null;
	
	private boolean endOfStream = false;
		
	public ParticipantRTPStream (Format format)
	{

		queue = new Vector<byte[]>();
		this.format = format;
		sync = new Object();
	}
	
	/* (non-Javadoc)
	 * @see javax.media.protocol.PushBufferStream#getFormat()
	 */
	public Format getFormat() {
		return format;
	}

	
	public void addPacket(byte [] packet) throws IOException, NotAValidHeaderException
	{
		/*Buffer buff = new Buffer();
		readRTPHeader(packet, buff);
		*/
		synchronized(sync)
		{
			queue.add(packet);
			if (queue.size()==1)
			{
				sync.notify();
				//nextHeader = new RTPHeader(packet);
			}
		}
	}
		
	private void readRTPPacket(byte [] packet, Buffer buff)
		throws NotAValidHeaderException
	{
		RTPHeader header = new RTPHeader(packet);
		int flags = Buffer.FLAG_RELATIVE_TIME | Buffer.FLAG_RTP_TIME;
		if (header.marker)
			flags = flags | Buffer.FLAG_RTP_MARKER;
		buff.setFlags(flags);
		buff.setHeader(header);
		buff.setFormat(format);
		buff.setTimeStamp(header.timestamp);
		int mediaDataSize = packet.length - header.totalHeaderSize;
		byte[] mediaData = new byte[mediaDataSize];
		System.arraycopy(packet, header.totalHeaderSize, mediaData, 0, mediaDataSize);
		buff.setData(mediaData);
		buff.setOffset(0);
		buff.setLength(mediaData.length);
	}
	
	
	public void close()
	{
		try {
			addPacket(new byte[1]);
		} catch (IOException e)
		{
			
		} catch (NotAValidHeaderException e)
		{
			
		}
	}


	public void read(Buffer buffer) throws IOException {
		synchronized (sync)
		{
			try {
				while (queue.isEmpty() && !endOfStream )
				{
					sync.wait();
				}
				if (!queue.isEmpty() || !endOfStream )
				{
					readRTPPacket(queue.remove(0), buffer);
					if ((buffer.getFlags() & Buffer.FLAG_SILENCE)!=0)
					{
						System.err.println("just a test");
					}
				}
			}
			catch (NotAValidHeaderException ex)
			{
				endOfStream = true;
				buffer.setEOM(true);
			}
			catch (InterruptedException e)
			{
				
			}
		}
	}


	public boolean willReadBlock() {
		boolean willBlock = false;
		synchronized(sync)
		{
			if (queue.isEmpty())
			{
				willBlock = true;
			}
		}
		return willBlock;
	}
	
	@Override
	public boolean endOfStream()
	{
		return endOfStream;
	}
}
